package com.fanxuankai.boot.mqbroker.producer;

import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.producer.autoconfigure.MqBrokerProducerProperties;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.PeriodicTrigger;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 定时任务配置
 *
 * @author fanxuankai
 */
public class ProducerTaskConfigurer implements SchedulingConfigurer {
    @Resource
    private MsgSendService service;
    @Resource
    private MqBrokerProperties brokerProperties;
    @Resource
    private MqBrokerProducerProperties properties;
    @Resource
    private MqProducer<MsgSend> mqProducer;
    @Resource
    private MsgSendMapper mapper;

    @Override
    public void configureTasks(@NonNull ScheduledTaskRegistrar scheduledTaskRegistrar) {
        if (properties.getDelayed().isEnabled()) {
            produceDelayedMsg(scheduledTaskRegistrar);
        }
        if (properties.getCompensation().isEnabled()) {
            compensateMsg(scheduledTaskRegistrar);
        }
        if (properties.getRetry().isEnabled()) {
            compensateLostMsg(scheduledTaskRegistrar);
        }
    }

    private void produceDelayedMsg(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.addTriggerTask(() -> {
            while (true) {
                List<MsgSend> records = service.pullDelayedData();
                if (records.isEmpty()) {
                    return;
                }
                records.forEach(msg -> {
                    if (service.lock(msg.getId(), Collections.emptyList())) {
                        mqProducer.produce(msg);
                    }
                });
            }
        }, triggerContext -> new PeriodicTrigger(properties.getDelayed().getInterval())
                .nextExecutionTime(triggerContext));
    }

    private void compensateMsg(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.addTriggerTask(() -> {
            while (true) {
                List<MsgSend> records = service.pullData();
                if (records.isEmpty()) {
                    return;
                }
                records.forEach(msg -> {
                    if (service.lock(msg.getId(), Collections.emptyList())) {
                        mqProducer.produce(msg);
                    }
                });
            }
        }, triggerContext -> new PeriodicTrigger(properties.getCompensation().getInterval())
                .nextExecutionTime(triggerContext));
        scheduledTaskRegistrar.addTriggerTask(() -> service.publisherCallbackTimeout(),
                triggerContext -> new PeriodicTrigger(properties.getCallbackTimeout())
                        .nextExecutionTime(triggerContext));
    }

    private void compensateLostMsg(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        Map<String, Set<String>> externalEventMap = getExternalEventMap();
        MqBrokerProducerProperties.Retry retry = properties.getRetry();
        long timeout = retry.getTimeout() / 1000;
        Set<String> externalEventNames = getExternalEventNames();
        scheduledTaskRegistrar.addTriggerTask(() -> {
            while (true) {
                List<Long> lost = mapper.lost(timeout, retry.getLimit(), externalEventNames);
                if (lost.isEmpty()) {
                    break;
                }
                service.compensate(lost);
            }
            if (!externalEventMap.isEmpty()) {
                for (Map.Entry<String, Set<String>> entry : externalEventMap.entrySet()) {
                    while (true) {
                        List<Long> lostExternal = mapper.lostExternal(timeout, retry.getLimit(), entry.getKey(), entry.getValue());
                        if (lostExternal.isEmpty()) {
                            return;
                        }
                        service.compensate(lostExternal);
                    }
                }
            }
        }, triggerContext -> new PeriodicTrigger(retry.getTimeout()).nextExecutionTime(triggerContext));
    }

    private Set<String> getExternalEventNames() {
        return brokerProperties.getExternalEvents()
                .stream()
                .map(MqBrokerProducerProperties.ExternalEvent::getName)
                .collect(Collectors.toSet());
    }

    private Map<String, Set<String>> getExternalEventMap() {
        return brokerProperties.getExternalEvents()
                .stream()
                .filter(externalEvent -> externalEvent.getSchema() != null)
                .collect(Collectors.groupingBy(MqBrokerProducerProperties.ExternalEvent::getSchema,
                        Collectors.mapping(MqBrokerProducerProperties.ExternalEvent::getName, Collectors.toSet())));
    }
}
